package com.freez.spark.core.demo;

import com.freez.spark.tool.TupleUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import scala.Tuple2;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: GoodsTotal.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:18
 * Description: 计算某一时间段中某商品的销售总额
 */
public class GoodsTotal {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        //集群
        //conf.setAppName("SparkDemo-GoodsTotal").setMaster("spark://slaveNode2:7077");
        conf.setAppName("SparkDemo-GoodsTotal").setMaster("spark://slaveNode2:7077");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> ordersRDD = jsc.textFile("./src/main/resources/data/core/demo/order.txt");
        JavaRDD<String> priceRDD = jsc.textFile("./src/main/resources/data/core/demo/product.txt");
        //处理data
        JavaPairRDD<Integer, Integer> orders = ordersRDD.mapToPair(s -> {
            String[] split = s.split(",");
            return TupleUtils.T2(Integer.parseInt(split[2]), Integer.parseInt(split[3]));
        });
        JavaPairRDD<Integer, Integer> price = priceRDD.mapToPair(s -> {
            String[] split = s.split(",");
            return TupleUtils.T2(Integer.parseInt(split[0]), Integer.parseInt(split[2]));
        });
        JavaPairRDD<Integer, String> name = priceRDD.mapToPair(s -> {
            String[] split = s.split(",");
            return TupleUtils.T2(Integer.parseInt(split[0]), split[1]);
        });
        //计算销量
        JavaPairRDD<Integer, Integer> totalNum = orders.reduceByKey((v1, v2) -> (v1 + v2));
        //
        JavaPairRDD<Integer, Tuple2<Integer, Optional<Integer>>> temp = price.leftOuterJoin(totalNum);
        //计算总额
        JavaPairRDD<Integer, Integer> totalMoneyRDD = temp.mapToPair(s -> {
            Integer singlePrice = s._2._1;
            if (s._2._2.isPresent()) {
                Integer totalMoney = singlePrice * s._2._2.get();
                return TupleUtils.T2(s._1, totalMoney);
            }
            return TupleUtils.T2(s._1, 0);
        });
        //将消费总额与商品对应
        JavaPairRDD<Integer, Tuple2<String, Optional<Integer>>> temp2 = name.leftOuterJoin(totalMoneyRDD);
        temp2.sortByKey().mapToPair(s -> {
            if (s._2._2.isPresent()) {
                Integer sell = s._2._2.get();
                return TupleUtils.T2(s._2._1, sell);
            }
            return TupleUtils.T2(s._2._1, 0);
        }).foreach(s -> System.out.println(s));
        jsc.close();
    }
}